package com.qianbao.finance.sync.modules.elasticsearch.client;

import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.xpack.client.PreBuiltXPackTransportClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.net.InetAddress;


@Component
public class ElasticsearchClient implements DisposableBean {

    private static final Logger logger = LoggerFactory.getLogger(ElasticsearchClient.class);

    private TransportClient transportClient;

    private RestHighLevelClient restHighLevelClient;

    private RestClient restClient;

    @Value("${elasticsearch.cluster.name}")
    private String clusterName;
    @Value("${elasticsearch.host}")
    private String host;
    @Value("${elasticsearch.port}")
    private String port;
    @Value("${elasticsearch.http.port}")
    private String httpPort;
    @Value("${elasticsearch.user}")
    private String user;
    @Value("${elasticsearch.passwd}")
    private String passwd;

    //@Bean
    public TransportClient getTransportClient() throws Exception {
        Settings settings = Settings.builder().put("cluster.name", clusterName)
                .put("client.transport.sniff", true)
                .put("xpack.security.user",user+":"+passwd)
                .build();
        String hostpart[] = host.split(",");
        String portpart[] = port.split(",");
        transportClient = new PreBuiltXPackTransportClient(settings)
                .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(hostpart[0]), Integer.valueOf(portpart[0])))
                .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(hostpart[1]),Integer.valueOf(portpart[1])))
                .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(hostpart[2]),Integer.valueOf(portpart[2])));
        logger.info("elasticsearch transportClient 连接成功");
        return transportClient;
    }

    @Bean
    public RestHighLevelClient getRestHighLevelClient() {
        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(getRestClient());
        return restHighLevelClient;
    }

    @Bean
    public RestClient getRestClient() {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(user, passwd));

        String hostpart[] = host.split(",");
        String httpPortpart[] = httpPort.split(",");
        HttpHost host1= new HttpHost(hostpart[0], Integer.parseInt(httpPortpart[0]));
        HttpHost host2= new HttpHost(hostpart[1], Integer.parseInt(httpPortpart[1]));
        HttpHost host3= new HttpHost(hostpart[2], Integer.parseInt(httpPortpart[2]));
        RestClientBuilder builder = RestClient.builder(host1,host2,host3)
                .setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider))
                .setFailureListener(new RestClient.FailureListener() {
                    @Override
                    public void onFailure(HttpHost host) {
                        logger.error("连接到es server出错！host-->"+host.getHostName());
                    }});
        return builder.build();
    }

    @Override
    public void destroy() throws Exception {
        if (transportClient != null) {
            transportClient.close();
        }
    }
}
